package ex.tableapi;

import ex.datastream.functions.userDefinedFunction.ScalarFunc;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 kafka作为数据源,查询(使用自定义MyFloor函数)数据并输出到控制台
 */
public class KafkaDemo2 extends ApiFrame {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        System.out.println("KafkaDemo2");
        //kafkas输入
       tEnv.executeSql("CREATE TABLE transactions (" +
                "    account_id  BIGINT," +
                "    amount      BIGINT," +
                "    transaction_time TIMESTAMP(3)," +
                "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND" +
                ") WITH (" +
                "    'connector' = 'kafka'," +
                "    'topic'     = 'test2'," +
                "    'properties.bootstrap.servers' = '192.168.1.6:9092'," +
                "    'properties.group.id' = 'testGroup'," +
                "    'scan.startup.mode' = 'earliest-offset'," +
                "    'format'    = 'csv'" +
                ")");


        Table source = tEnv.from("transactions");

        //定义输出
        String createInDDL =
                        "CREATE TABLE printOutTable (" +
                        "  `account_id` BIGINT," +
                        "  `log_ts` TIMESTAMP(3)," +
                        "  `amount` BIGINT" +
                        ") WITH (" +
                        "  'connector' = 'print'" +
                        ")";
        tEnv.executeSql(createInDDL);

        //转换
        Table result=source.select(
                $("account_id"), call(ScalarFunc.class, $("transaction_time")).as("log_ts"),
                $("amount"))
                .groupBy($("account_id"), $("log_ts"))
                .select(
                        $("account_id"),
                        $("log_ts"),
                        $("amount").sum().as("amount"));
        //查看转换后格式
        result.printSchema();
        //转换结果输出到控制台
        result.executeInsert("printOutTable");
    }
}
